set hive.exec.dynamic.partition=true;            --  动态分区
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=90;         --  每天生成 60 个分区
set hive.exec.max.dynamic.partitions.pernode=90; --  每天生成 60 个分区

-- 操作数据合并
with barscans_incre_tmp as (
    select
        "taking" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_taking
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "warehousing" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_warehousing
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "siteSend" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_sitesend
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "siteArrival" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_sitearrival
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "unloading" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_unloading
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "bagging" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from jms_ods.tab_barscan_bagging
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "centerSend" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_centersend
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "centerArrival" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_centerarrival
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "loading" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from  jms_ods.tab_barscan_loading
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "deliver" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from jms_ods.tab_barscan_deliver
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        "sign" as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,delivertime as scan_time
        ,null as recordid
        ,null as packagecode
        ,nextstationcode
        ,null as nextstation
        ,destination
        ,scanusercode
        ,null as scanuser
        ,source
        ,cast(realweight as string) as weight
        ,null as expresstype
        ,null as returnflag
        ,null as elescaleflag
        ,null as pistolid
        ,null as transfercode
        ,null as sendcustomer
        ,null as subbillcode
        ,'0' as CHECKFLAG
    from jms_ods.tab_barscan_sign
    where dt =  '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        case scantype
        when '出库扫描' then 'outStore'
        when '快件取出扫描' then 'expressOut'
        when '入库扫描' then 'inStore'
        else 'collect_others'
        end as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from jms_ods.tab_barscan_collect_back
    where dt = '{{ execution_date | cst_ds }}'
    and (billcode like "JT%" or billcode like "55%")

    union all

    select
        case scantype
        when '拆包扫描' then 'unbagging'
        when '封发车扫描' then 'carsend'
        when '到解车扫描' then 'uncarArrival'
        when '留仓件入仓' then 'twiceWarehousing'
        else 'other_others'
        end as scan_type
        ,billcode
        ,scansitecode as network_code
        ,null as network_type
        ,scantime as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,cast(weight as string)  as  weight
        ,expresstype
        ,cast(returnflag as string)  as returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,case CHECKFLAG when '7' then '1' else     '0'  end  as check_flag
    from jms_ods.tab_barscan_other
    where dt = '{{ execution_date | cst_ds }}'
    and scansitecode is not null
    and (billcode like "JT%" or billcode like "55%")
    and scansitecode is not null

    union all

    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,sendcustomer
        ,subbillcode
        ,check_flag
    from jms_dwd.dwd_s03_barscan_operations_hist_dt
    where dt = '2049-12-31'
    and date_format(scan_time, 'yyyy-MM-dd') >= date_add('{{ execution_date | cst_ds }}', -29)
    and date_format(scan_time, 'yyyy-MM-dd') <= '{{ execution_date | cst_ds }}'
)
,hist_tmp as (
    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,is_sign
        ,input_time
        ,operation_dt
        ,sendcustomer
        ,subbillcode
        ,roadsection_id
        ,planned_arrival_time
        ,actual_arrival_time
        ,planned_departure_time
        ,actual_departure_time
        ,latest_arrival_time
        ,table_type
        ,vehicleline_code
        ,vehicleline_name
        ,shipment_name
        ,is_transfer
        ,check_flag
        ,dt
    from jms_dwd.dwd_s03_barscan_operations_hist_dt
    where dt >= date_add('{{ execution_date | cst_ds }}', -29)
    and dt <= '{{ execution_date | cst_ds }}'
    and operation_dt <> '{{ execution_date | cst_ds }}'
)
-- DESC：操作合并数据与运单增量表以 waybill_no 关联，获取运单创建时间
-- 关联不到的运单及运单创建时间获取不到，则赋值为'2049-12-31 23:59:59'
,incre_data_tmp as (
    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,cast(scan_time as string) as scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,is_sign
        ,input_time
        ,'{{ execution_date | cst_ds }}' as operation_dt
        ,date_format(input_time,'yyyy-MM-dd') as dt
        ,sendcustomer
        ,subbillcode
        ,check_flag
    from (
        select
            /*+ BROADCAST(t2) */
            t1.*,t2.is_sign
            ,cast(nvl(t2.input_time,'2049-12-31 23:59:59') as timestamp) as input_time
        from barscans_incre_tmp t1
        left join (
                select
                    waybill_no,input_time,is_sign
                from jms_ods.yl_oms_oms_waybill
                where dt between date_add('{{ execution_date | cst_ds }}', -29)
                and '{{ execution_date | cst_ds }}'
                and date_format(input_time,'yyyy-MM-dd') = dt
        ) t2
        on t1.billcode = t2.waybill_no
    ) tmp
)
,tab_stop as (
     select
         shipment_no
         ,network_code
         ,roadsection_id
         ,planned_arrival_time
         ,actual_arrival_time
         ,planned_departure_time
         ,actual_departure_time
         ,latest_arrival_time
         ,table_type
         ,sort_num
     from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt
     where dt >= date_add('{{ execution_date | cst_ds }}', -90)
     and dt <= date_add('{{ execution_date | cst_ds }}',7)
)
,tab_shipment as (
     select shipment_no,shipment_name,vehicleline_code,vehicleline_name,cnt
     from jms_dwd.dwd_tmsnew_shipment_union_base_dt
     where dt >= date_add('{{ execution_date | cst_ds }}', -90)
     and dt <= date_add('{{ execution_date | cst_ds }}',7)
)
,shipment_tmp as (
     SELECT
         shipment_no                   AS shipment_no             -- 转运任务单号
         ,shipment_name                AS shipment_name           -- 转运任务单号名称
         ,network_code                 AS network_code            -- 网点编码
         ,min(roadsection_id)          AS roadsection_id          -- 路段编号
         ,min(planned_arrival_time)    AS planned_arrival_time    -- 计划到达日期+时间
         ,min(actual_arrival_time)     AS actual_arrival_time     -- 实际到达时间
         ,min(planned_departure_time)  AS planned_departure_time  -- 计划离开日期+时间
         ,min(actual_departure_time)   AS actual_departure_time   -- 实际发车时间
         ,min(latest_arrival_time)     AS latest_arrival_time     -- 最晚到车时间+日期
         ,min(table_type)              AS table_type              -- 支干线类型
         ,min(vehicleline_code)        AS vehicleline_code        -- 参考车线编码
         ,min(vehicleline_name)        AS vehicleline_name        -- 参考车线名称
         ,if(min(sort_num) < min(cnt) and min(sort_num) > 1 ,1,0) AS is_transfer        -- 是否中转
     FROM (
           select
                /*+ BROADCAST(tab_shipment) */
               t1.shipment_no
               ,t2.shipment_name
               ,network_code
               ,roadsection_id
               ,planned_arrival_time
               ,actual_arrival_time
               ,planned_departure_time
               ,actual_departure_time
               ,latest_arrival_time
               ,table_type
               ,vehicleline_code
               ,vehicleline_name
               ,sort_num
               ,cnt
           from tab_stop t1
           left join tab_shipment t2
           on t1.shipment_no = t2.shipment_no
     ) as  stops
     GROUP BY shipment_no, network_code, shipment_name
)
,hist_data_join_tms (
    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,is_sign
        ,input_time
        ,operation_dt
        ,sendcustomer
        ,subbillcode
        ,check_flag
        ,dt
    from hist_tmp
    where scan_type in ('siteSend','siteArrival','unloading','centerSend','centerArrival','loading')

    union all

    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,is_sign
        ,input_time
        ,operation_dt
        ,sendcustomer
        ,subbillcode
        ,check_flag
        ,dt
    from incre_data_tmp
)
,incre_tms_data_tmp as (
  select
          /*+ BROADCAST(shipment_tmp) */
          t1.scan_type
         ,t1.billcode
         ,t1.network_code
         ,t1.network_type
         ,t1.scan_time
         ,t1.recordid
         ,t1.packagecode
         ,t1.nextstationcode
         ,t1.nextstation
         ,t1.destination
         ,t1.scanusercode
         ,t1.scanuser
         ,t1.source
         ,t1.weight
         ,t1.expresstype
         ,t1.returnflag
         ,t1.elescaleflag
         ,t1.pistolid
         ,t1.transfercode
         ,t1.is_sign
         ,t1.input_time
         ,t1.operation_dt
         ,t1.dt
         ,t1.sendcustomer
         ,t1.subbillcode
         ,t2.roadsection_id
         ,t2.planned_arrival_time
         ,t2.actual_arrival_time
         ,t2.planned_departure_time
         ,t2.actual_departure_time
         ,t2.latest_arrival_time
         ,t2.table_type
         ,t2.vehicleline_code
         ,t2.vehicleline_name
         ,t2.shipment_name
         ,t2.is_transfer
         ,t1.check_flag
  from hist_data_join_tms t1
  left join shipment_tmp t2
  on if(t1.transfercode is null or t1.transfercode in ('','null','NULL'),t1.billcode,t1.transfercode) = t2.shipment_no
  and if(t1.network_code is null or t1.network_code in ('','null','NULL'),t1.billcode,t1.network_code) = t2.network_code
)

-- DESC: 将关联到运单创建时间的昨日操作数据 与 操作合并表近30天数据合并 重新回插入操作合并表
insert overwrite table jms_dwd.dwd_s03_barscan_operations_hist_dt partition(dt)
select
    scan_type
    ,billcode
    ,network_code
    ,network_type
    ,scan_time
    ,recordid
    ,packagecode
    ,nextstationcode
    ,nextstation
    ,destination
    ,scanusercode
    ,scanuser
    ,source
    ,weight
    ,expresstype
    ,returnflag
    ,elescaleflag
    ,pistolid
    ,transfercode
    ,is_sign
    ,input_time
    ,operation_dt
    ,sendcustomer
    ,subbillcode
    ,roadsection_id
    ,planned_arrival_time
    ,actual_arrival_time
    ,planned_departure_time
    ,actual_departure_time
    ,latest_arrival_time
    ,table_type
    ,vehicleline_code
    ,vehicleline_name
    ,shipment_name
    ,is_transfer
    ,check_flag
    ,dt
from (
    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,is_sign
        ,input_time
        ,operation_dt
        ,sendcustomer
        ,subbillcode
        ,roadsection_id
        ,planned_arrival_time
        ,actual_arrival_time
        ,planned_departure_time
        ,actual_departure_time
        ,latest_arrival_time
        ,table_type
        ,vehicleline_code
        ,vehicleline_name
        ,shipment_name
        ,is_transfer
        ,check_flag
        ,dt
    from incre_tms_data_tmp

    union all

    select
        scan_type
        ,billcode
        ,network_code
        ,network_type
        ,scan_time
        ,recordid
        ,packagecode
        ,nextstationcode
        ,nextstation
        ,destination
        ,scanusercode
        ,scanuser
        ,source
        ,weight
        ,expresstype
        ,returnflag
        ,elescaleflag
        ,pistolid
        ,transfercode
        ,is_sign
        ,input_time
        ,operation_dt
        ,sendcustomer
        ,subbillcode
        ,roadsection_id
        ,planned_arrival_time
        ,actual_arrival_time
        ,planned_departure_time
        ,actual_departure_time
        ,latest_arrival_time
        ,table_type
        ,vehicleline_code
        ,vehicleline_name
        ,shipment_name
        ,is_transfer
        ,check_flag
        ,dt
    from hist_tmp
    where scan_type not in ('siteSend','siteArrival','unloading','centerSend','centerArrival','loading')
) tmp
where dt >= date_add('{{ execution_date | cst_ds }}', -29)
distribute by dt, abs(hash(billcode)) % 300
;